Node / ReticulumProjects / MeshChatX.git / files / vendor / lxmfy / tests / test_recursive_cogs.py

Displaying Raw • Download

Tff7b72import T7ee787os

Tff7b72import T7ee787stat

Tff7b72import T7ee787tempfile

Tff7b72from T7ee787pathlib Tff7b72import Te6edf3Path

Tff7b72from T7ee787lxmfy Tff7b72import Te6edf3BotConfigTb4b4b4, Te6edf3LXMFBot

Tff7b72from T7ee787lxmfyT7ee787.T7ee787cogs_core Tff7b72import Te6edf3load_cogs_from_directory

Tff7b72def Td2a8fftest_cog_execution_depth_limitTb4b4b4(Te6edf3test_config_dirTb4b4b4)Tb4b4b4:

T8b949e"""Test that external cogs cannot trigger infinite recursive bot calls."""

T8b949e# This is a bit complex to test fully without a real bot instance,

T8b949e# but we can verify the threading and isolation logic.

Tff7b72with Te6edf3tempfileTff7b72.Td2a8ffTemporaryDirectoryTb4b4b4(Tb4b4b4) Tff7b72as Te6edf3temp_dirTb4b4b4:

Te6edf3temp_path Tff7b72= Te6edf3PathTb4b4b4(Te6edf3temp_dirTb4b4b4)

Te6edf3cogs_dir Tff7b72= Te6edf3temp_path Tff7b72/ Ta5d6ff"Ta5d6ffcogsTa5d6ff"

Te6edf3cogs_dirTff7b72.Td2a8ffmkdirTb4b4b4(Tb4b4b4)

T8b949e# Create a cog that tries to call the bot recursively

T8b949e# (simulated by executing another command)

Te6edf3script_path Tff7b72= Te6edf3cogs_dir Tff7b72/ Ta5d6ff"Ta5d6ffrecursive.shTa5d6ff"

Te6edf3script_content Tff7b72= Ta5d6ff"""Ta5d6ff#!/bin/bash

Ta5d6ffecho Ta5d6ff"Ta5d6ffCalling self...Ta5d6ff"

Ta5d6ff# In a real scenario, this might try to send an LXMF message back to the bot

Ta5d6ff# which would trigger another command execution.

Ta5d6ff"""

Te6edf3script_pathTff7b72.Td2a8ffwrite_textTb4b4b4(Te6edf3script_contentTb4b4b4)

Te6edf3osTff7b72.Td2a8ffchmodTb4b4b4(Te6edf3script_pathTb4b4b4, Te6edf3osTff7b72.Td2a8ffstatTb4b4b4(Te6edf3script_pathTb4b4b4)Tff7b72.Td2a8ffst_mode Tff7b72| Te6edf3statTff7b72.Td2a8ffS_IEXECTb4b4b4)

Te6edf3config Tff7b72= Te6edf3BotConfigTb4b4b4(

Te6edf3nameTff7b72=Ta5d6ff"Ta5d6ffDepthTestBotTa5d6ff"Tb4b4b4,

Te6edf3test_modeTff7b72=Tff7b72TrueTb4b4b4,

Te6edf3external_cogs_enabledTff7b72=Tff7b72TrueTb4b4b4,

Te6edf3external_cogs_timeoutTff7b72=T79c0ff2Tb4b4b4,

Te6edf3config_pathTff7b72=Tffa657strTb4b4b4(Te6edf3temp_pathTb4b4b4)Tb4b4b4,

Te6edf3storage_pathTff7b72=Tffa657strTb4b4b4(Te6edf3temp_path Tff7b72/ Ta5d6ff"Ta5d6ffstorageTa5d6ff"Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3bot Tff7b72= Te6edf3LXMFBotTb4b4b4(Tff7b72*Tff7b72*Te6edf3configTff7b72.Td2a8ff__dict__Tb4b4b4)

Te6edf3botTff7b72.Td2a8ffconfig_path Tff7b72= Tffa657strTb4b4b4(Te6edf3temp_pathTb4b4b4)

Te6edf3load_cogs_from_directoryTb4b4b4(Te6edf3botTb4b4b4)

Te6edf3cmd Tff7b72= Te6edf3botTff7b72.Td2a8ffcommandsTb4b4b4[Ta5d6ff"Ta5d6ffrecursiveTa5d6ff"Tb4b4b4]

T8b949e# Since we use threading and subprocess.run with timeout,

T8b949e# any actual recursive fork-bomb would be caught by the process limit

T8b949e# or the bot's thread pool executor limit.

Tff7b72assert Te6edf3cmdTff7b72.Td2a8ffthreaded Tff7b72is Tff7b72True

T8b949e# Verify the bot's thread pool is initialized

Tff7b72assert Te6edf3botTff7b72.Td2a8ffthread_poolTff7b72.Td2a8ff_max_workers Tff7b72== T79c0ff5